package org.example.consumer;

import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.example.bean.Alipay;
import org.example.bean.Logistics;
import org.example.bean.Stock;
import org.example.mapper.StockMapper;
import org.example.service.StockService;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @Auther:Liu
 * @Testname:ChangeStock
 * @Date:2025/8/24 15:37
 */
@Component
@Slf4j
public class StockConsumer {
    @Resource
    private StockService stockService;
    @Resource
    private StockMapper stockMapper;

    @SneakyThrows
    @Scheduled(initialDelay = 1000)
    public void updateStock() {
        log.info("启动");
        log.info("stockService:{}", stockService);
        log.info("stockMapper:{}", stockMapper);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("producer_group_c");
        consumer.setNamesrvAddr("192.168.10.88:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("Order", "alipay");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            log.info("物流收到信息");
            for (MessageExt msg : list) {
                try {
                    log.info("获取");
                    byte[] body = msg.getBody();
                    String json = new String(body, StandardCharsets.UTF_8);
                    Alipay alipay = JSONArray.parseObject(json, Alipay.class);
                    QueryWrapper<Stock> wrapper = new QueryWrapper<>();
                    wrapper.eq("book_id", alipay.getBookId());
                    log.info("id:{}", alipay.getBookId());
                    Stock one = stockService.getOne(wrapper);
                    log.info("one:{}", one);
                    UpdateWrapper<Stock> updateWrapper = new UpdateWrapper<>();
                    updateWrapper.eq("book_id", alipay.getBookId());
                    updateWrapper.set("number", one.getNumber() - alipay.getNumber());
                    stockService.update(updateWrapper);
                } catch (Exception e) {
                    log.error("<UNK>", e);
                    log.info("异常");
                    log.info("stockService:{}", stockService);
                    throw new RuntimeException(e);
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }
}